package com.sunwayworld.basemodule.common.asynctask.util;

import com.sunwayworld.basemodule.common.asynctask.bean.LimsAsynctaskBean;
import com.sunwayworld.basemodule.common.asynctask.bean.LimsAsynctaskScheduleBean;
import com.sunwayworld.basemodule.common.asynctask.bean.LimsAsynctaskServerBean;
import com.sunwayworld.basemodule.common.asynctask.comsumer.AsyncTaskConsumer;
import com.sunwayworld.basemodule.common.asynctask.config.AsyncTaskExecutorConfig;
import com.sunwayworld.basemodule.common.asynctask.config.AsyncTaskStatusEnum;
import com.sunwayworld.basemodule.common.asynctask.config.LimsAsyncTaskExecutor;
import com.sunwayworld.basemodule.common.asynctask.service.LimsAsynctaskScheduleService;
import com.sunwayworld.basemodule.common.asynctask.service.LimsAsynctaskServerService;
import com.sunwayworld.basemodule.common.asynctask.service.LimsAsynctaskService;
import com.sunwayworld.basemodule.common.eptstorage.db.DynamicBeanHanderService;
import com.sunwayworld.framework.context.concurrent.GikamConcurrentLocker;
import com.sunwayworld.framework.database.sql.Order;
import com.sunwayworld.framework.exception.TaskException;
import com.sunwayworld.framework.mybatis.mapper.MatchPattern;
import com.sunwayworld.framework.mybatis.mapper.SearchFilter;
import com.sunwayworld.framework.utils.ObjectUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.support.TransactionSynchronizationAdapter;
import org.springframework.transaction.support.TransactionSynchronizationManager;

import java.util.*;
import java.util.concurrent.*;

@Service
public class AsyncTaskExecutorHelper {
    private static final Logger logger = LogManager.getLogger(AsyncTaskExecutorHelper.class);
    public static final String AsyncTaskExecutorNamePrefix = "limsAsyncTaskExecutor";
    @Autowired(required = false)
    private AsyncTaskExecutorConfig taskExecutorConfig;
    @Autowired
    private DynamicBeanHanderService dynamicBeanHanderService;
    @Autowired
    private LimsAsynctaskService asynctaskService;
    @Autowired
    private LimsAsynctaskScheduleService asynctaskScheduleService;
    @Autowired
    private LimsAsynctaskServerService asynctaskServerService;
    @Autowired
    private AsyncTaskConsumer taskScheduleConsumer;

    /**
     节点启动时，注入线程池，线程池配置信息来自于LimsAsynctaskScheduleBean
     */
    public void register(){
        if(taskExecutorConfig==null){
            return;
        }
        boolean enable = taskExecutorConfig.getEnable();
        String currentServerId = taskExecutorConfig.getCurrentServerId();
        if(!enable || ObjectUtils.isEmpty(currentServerId)){
            return;
        }

        LimsAsynctaskServerBean asynctaskServerBean = asynctaskServerService.selectByIdIfPresent(currentServerId);
        if(ObjectUtils.isEmpty(asynctaskServerBean)){
            logger.warn("currentServerId={},not find LimsAsynctaskServer",currentServerId);
            return;
        }
        if(!"1".equals(asynctaskServerBean.getEnable())){
            logger.warn("currentServerId={},LimsAsynctaskServer not enable",currentServerId);
            return;
        }
        List<LimsAsynctaskScheduleBean> asynctaskScheduleBeans = asynctaskScheduleService.selectListByFilter(SearchFilter.instance().match("serverId", currentServerId).filter(MatchPattern.SEQ)
                                                                    .match("enable","1").filter(MatchPattern.SEQ));
        if(ObjectUtils.isEmpty(asynctaskScheduleBeans)){
            logger.warn("currentServerId={},not find LimsAsynctaskSchedule",currentServerId);
            return;
        }

        logger.info("start register LimsAsynctaskSchedule...");
        for (LimsAsynctaskScheduleBean executor : asynctaskScheduleBeans) {
            String executorId = executor.getExecutorId();
            String name = executor.getName();
            Long corePoolSize = executor.getCorePoolSize();
            Long maxPoolSize = executor.getMaxPoolSize();
            Long queueCapacity = executor.getQueueCapacity();
            Objects.requireNonNull(executorId, "AsynctaskSchedule executorId is null");
            Objects.requireNonNull(name, "AsynctaskSchedule name is null");
            Objects.requireNonNull(corePoolSize, "AsynctaskSchedule corePoolSize is null");
            Objects.requireNonNull(maxPoolSize, "AsynctaskSchedule maxPoolSize is null");
            Objects.requireNonNull(queueCapacity, "AsynctaskSchedule queueCapacity is null");

            Map<String, Object> pvs = new HashMap<>();
            pvs.put("corePoolSize",corePoolSize);
            pvs.put("maxPoolSize",maxPoolSize);
            pvs.put("queueCapacity",queueCapacity);
            String beanId = getExecutorBeanId(executorId);
            pvs.put("name",beanId);

            try{
                dynamicBeanHanderService.register(beanId,LimsAsyncTaskExecutor.class,beanId,corePoolSize,maxPoolSize,queueCapacity);
                try {
                    getByExecutorName(executorId);
                } catch (Exception throwables) {
                    logger.error("getBean={} is fail={}",beanId,throwables);
                    dynamicBeanHanderService.destory(beanId);
                }
            }catch (Throwable s){
                logger.error("register={} is fail={}",beanId,s);
            }
        }
        logger.info("end register LimsAsynctaskSchedule...");
    }

    private String getExecutorBeanId(String executorName){
        return AsyncTaskExecutorNamePrefix+executorName;
    }

    //根据executor名称获取线程池
    public LimsAsyncTaskExecutor getByExecutorName(String executorName){
        if(ObjectUtils.isEmpty(executorName)){
            return null;
        }
        String beanId = getExecutorBeanId(executorName);
        Object serviceBean = dynamicBeanHanderService.getBean(beanId);
        if(serviceBean==null){
            logger.error("getBean={} is not found",beanId);
            return null;
        }
        return (LimsAsyncTaskExecutor) serviceBean;
    }

    public  <T> Future<T> submit(String executorName,Callable<T> task) {
        LimsAsyncTaskExecutor executor = getByExecutorName(executorName);
        return executor.submit(task);
    }

    public  void submit(String executorName,Runnable task) {
        LimsAsyncTaskExecutor executor = getByExecutorName(executorName);
        executor.submit(task);
    }

    public  <T> T submit(String executorName,Callable<T> task, long timeoutMilliSeconds) {
        LimsAsyncTaskExecutor executor = getByExecutorName(executorName);
        Future<T> future = executor.submit(task);
        try {
            return future.get(timeoutMilliSeconds, TimeUnit.MILLISECONDS);
        } catch (Exception var7) {
            try {
                future.cancel(true);
            } catch (Exception var6) {
            }
            throw new TaskException(task.toString(), var7);
        }
    }

    public  void submitAfterCommitTransaction(String executorName,Runnable task) {
        TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {
            public void afterCommit() {
                LimsAsyncTaskExecutor executor = getByExecutorName(executorName);
                executor.submit(task);
            }
        });
    }

    public boolean hasRegister(String executorName){
        LimsAsyncTaskExecutor executor = getByExecutorName(executorName);
        return executor!=null;
    }

    //查询还未分配或者未加入到线程池中的，或已加入但是未执行的
    public List<LimsAsynctaskBean> getWaitingAsyncTask(List<String> currentServerIds){
        Integer execMaxCount = 1;
        List<LimsAsynctaskBean> hasServerServerTasks = asynctaskService.selectListByFilter(SearchFilter.instance().match("serverId", currentServerIds).filter(MatchPattern.OR)
                .match("execStatus", Arrays.asList(AsyncTaskStatusEnum.hasAssign.status,AsyncTaskStatusEnum.executeFail.status)).filter(MatchPattern.OR)
                .match("execCount", execMaxCount).filter(MatchPattern.NL), Order.asc("createdTime")); //小于最大次数

        List<LimsAsynctaskBean> noneServerTasks = asynctaskService.selectListByFilter(SearchFilter.instance().match("serverId",null).filter(MatchPattern.SEQ)
                .match("execStatus", Arrays.asList(AsyncTaskStatusEnum.pendingAssign.status)).filter(MatchPattern.OR)
                .match("execCount", execMaxCount).filter(MatchPattern.NL), Order.asc("createdTime")); //小于最大次数
        List<LimsAsynctaskBean> list = new ArrayList<>();
        if(!ObjectUtils.isEmpty(hasServerServerTasks)){
            list.addAll(hasServerServerTasks);
        }
        if(!ObjectUtils.isEmpty(noneServerTasks)){
            list.addAll(noneServerTasks);
        }
        return list;
    }

    //求取任务的排队位数
    public void updateWaitAsyncTaskWaitNum(List<String> currentServerIds){
     /*   if(ObjectUtils.isEmpty(currentServerIds)){
            return;
        }
        List<LimsAsynctaskBean> waitingAsyncTask = getWaitingAsyncTask(currentServerIds);
        if(ObjectUtils.isEmpty(waitingAsyncTask)){
            return;
        }
        List<String> executorNames = waitingAsyncTask.stream().map(f -> f.getExecutorName()).distinct().collect(Collectors.toList());
        Map<String, Pair<Integer,Integer>> taskMap = new HashMap<>();
        Map<String,Integer> executorMap = new HashMap<>();
        for (String executorName : executorNames) {
            Integer taskNumTotal = 0;
            for (String currentServerId : currentServerIds) {
                List<AsyncTaskExecutorConfig.Executor> executors = asynctaskService.selectExecutorList();
                Optional<AsyncTaskExecutorConfig.Executor> optional = executors.stream().filter(g -> currentServerId.equals(g.getExt$().get("id")) && executorName.equals(g.getName())).findFirst();
                if(optional.isPresent()){
                    AsyncTaskExecutorConfig.Executor executor = optional.get();
                    Pair<Integer, Integer> monitorTaskInfo = new Pair<>(executor.getActiveCount(),executor.getWaitQueueSize());
                    taskMap.put(executorName+"::"+currentServerId,monitorTaskInfo);
                    taskNumTotal = taskNumTotal + monitorTaskInfo.getKey()+monitorTaskInfo.getValue();
                }
            }
            executorMap.put(executorName,taskNumTotal);
        }

        //所以排队位数是所在节点的executor.activecount++  hasAssign joinedQueue executeFail
        //如果任务是其他状态，因为任务还未分配到某个节点，则排队位数是所有节点上的指定executor的taskNum总和，再++  pendingAssign
        List<LimsAsynctaskBean> joinedQueueTasks = waitingAsyncTask.stream().filter(f -> AsyncTaskStatusEnum.hasAssign.status.equals(f.getExecStatus())
                                                                            || AsyncTaskStatusEnum.joinedQueue.status.equals(f.getExecStatus())
                                                                            || AsyncTaskStatusEnum.executeFail.status.equals(f.getExecStatus())).collect(Collectors.toList());
        for (String currentServerId : currentServerIds) {
            List<LimsAsynctaskBean> cur_joinedQueueTasks = joinedQueueTasks.stream().filter(f -> f.getServerId().equals(currentServerId)).collect(Collectors.toList());
            List<String> exeuctorNames = cur_joinedQueueTasks.stream().map(f -> f.getExecutorName()).distinct().collect(Collectors.toList());
            for (String exeuctorName : exeuctorNames) {
                Pair<Integer,Integer> pair = taskMap.get(exeuctorName + "::" + currentServerId);
                if(pair!=null){
                    Integer activeCount = pair.getKey();
                    List<LimsAsynctaskBean> beans = cur_joinedQueueTasks.stream().filter(f -> f.getExecutorName().equals(exeuctorName)).collect(Collectors.toList());
                    for (int i = 0; i < beans.size(); i++) {
                        beans.get(i).setQueueProgress((long) (i+activeCount+1));
                    }
                }
            }
        }

        List<LimsAsynctaskBean> otherTasks = waitingAsyncTask.stream().filter(f -> AsyncTaskStatusEnum.pendingAssign.status.equals(f.getExecStatus())).collect(Collectors.toList());
        for (String executorName : executorNames) {
            List<LimsAsynctaskBean> beans = otherTasks.stream().filter(f -> f.getExecutorName().equals(executorName)).collect(Collectors.toList());
            Integer num = executorMap.get(executorName);
            for (int i = 0; i < beans.size(); i++) {
                beans.get(i).setQueueProgress((long) (i+1+num));
            }
        }

        asynctaskService.getDao().update(waitingAsyncTask,"QueueProgress");*/
    }

    private static ScheduledExecutorService fixTaskScheduleGetTask;

    /**
     * 启动定时任务，获取任务线程
     */
    public void startFixTaskScheduleGetTask(){
        if(taskExecutorConfig==null){
            return;
        }
        boolean enable = taskExecutorConfig.getEnable();
        String currentServerId = taskExecutorConfig.getCurrentServerId();
        if(!enable || ObjectUtils.isEmpty(currentServerId)){
            return;
        }
        LimsAsynctaskServerBean asynctaskServerBean = asynctaskServerService.selectByIdIfPresent(currentServerId);
        if(ObjectUtils.isEmpty(asynctaskServerBean)){
            logger.warn("currentServerId={},not find LimsAsynctaskServer",currentServerId);
            return;
        }
        if(!"1".equals(asynctaskServerBean.getEnable())){
            logger.warn("currentServerId={},LimsAsynctaskServer not enable",currentServerId);
            return;
        }
        List<LimsAsynctaskScheduleBean> asynctaskScheduleBeans = asynctaskScheduleService.selectListByFilter(SearchFilter.instance().match("serverId", currentServerId).filter(MatchPattern.SEQ)
                .match("enable","1").filter(MatchPattern.SEQ));
        if(ObjectUtils.isEmpty(asynctaskScheduleBeans)){
            logger.warn("currentServerId={},not find LimsAsynctaskSchedule",currentServerId);
            return;
        }

        if(fixTaskScheduleGetTask==null){
            fixTaskScheduleGetTask = Executors.newScheduledThreadPool(1);
            Long scheduleGetTaskSecond = taskExecutorConfig.getScheduleGetTaskSecond();
            if(scheduleGetTaskSecond==null){
                scheduleGetTaskSecond = 60L;
            }
            fixTaskScheduleGetTask.scheduleWithFixedDelay (() -> {
                GikamConcurrentLocker.isolatedRun("startFixTaskScheduleGetTask",()->{
                    try {
                        asynctaskService.initialTaskOnDeadLives();

                        taskScheduleConsumer.searchTaskAndExecute();
                    }catch (Throwable s){
                        logger.error("FixTaskScheduleGetTask error : {}",s);
                    }
                });
            }, 60, scheduleGetTaskSecond, TimeUnit.SECONDS);
        }
    }
}
